Spring Cloud Stream 体系及原理介绍

2023-12-28 23:56| 来源: 网络整理| 查看: 265

头图.png

作者 | 洛夜来源 | 阿里巴巴云原生公众号

Spring Cloud Stream在 Spring Cloud 体系内用于构建高度可扩展的基于事件驱动的微服务，其目的是为了简化消息在 Spring Cloud 应用程序中的开发。

Spring Cloud Stream (后面以 SCS 代替 Spring Cloud Stream) 本身内容很多，而且它还有很多外部的依赖，想要熟悉 SCS，必须要先了解 Spring Messaging 和 Spring Integration 这两个项目，接下来，文章将围绕以下三点进行展开：

什么是 Spring Messaging什么是 Spring Integration什么是 SCS 体系及其原理

本文配套可交互教程已登录阿里云知行动手实验室，PC 端登录 start.aliyun.com_ _在浏览器中立即体验。

Spring Messaging

Spring Messaging 是 Spring Framework 中的一个模块，其作用就是统一消息的编程模型。

比如消息 Messaging 对应的模型就包括一个消息体 Payload 和消息头 Header：

package org.springframework.messaging; public interface Message { T getPayload(); MessageHeaders getHeaders(); }消息通道 MessageChannel 用于接收消息，调用send方法可以将消息发送至该消息通道中：

@FunctionalInterface public interface MessageChannel { long INDEFINITE_TIMEOUT = -1; default boolean send(Message message) { return send(message, INDEFINITE_TIMEOUT); } boolean send(Message message, long timeout); }消息通道里的消息如何被消费呢？由消息通道的子接口可订阅的消息通道SubscribableChannel实现，被MessageHandler消息处理器所订阅：public interface SubscribableChannel extends MessageChannel { boolean subscribe(MessageHandler handler); boolean unsubscribe(MessageHandler handler); }由MessageHandler真正地消费/处理消息：@FunctionalInterface public interface MessageHandler { void handleMessage(Message message) throws MessagingException; }Spring Messaging 内部在消息模型的基础上衍生出了其它的一些功能，如：消息接收参数及返回值处理：消息接收参数处理器HandlerMethodArgumentResolver配合@Header, @Payload等注解使用；消息接收后的返回值处理器HandlerMethodReturnValueHandler配合@SendTo注解使用；消息体内容转换器MessageConverter；统一抽象的消息发送模板AbstractMessageSendingTemplate；消息通道拦截器ChannelInterceptor；Spring Integration

Spring Integration 提供了 Spring 编程模型的扩展用来支持企业集成模式(Enterprise Integration Patterns)，是对 Spring Messaging 的扩展。

它提出了不少新的概念，包括消息路由MessageRoute、消息分发MessageDispatcher、消息过滤Filter、消息转换Transformer、消息聚合Aggregator、消息分割Splitter等等。同时还提供了MessageChannel和MessageHandler的实现，分别包括 DirectChannel、ExecutorChannel、PublishSubscribeChannel和MessageFilter、ServiceActivatingHandler、MethodInvokingSplitter 等内容。

这里为大家介绍几种消息的处理方式：消息的分割：

消息的聚合：

消息的过滤：

消息的分发：

接下来，我们以一个最简单的例子来尝试一下 Spring Integration。

这段代码解释为：

SubscribableChannel messageChannel =new DirectChannel(); // 1 messageChannel.subscribe(msg-> { // 2 System.out.println("receive: " +msg.getPayload()); }); messageChannel.send(MessageBuilder.withPayload("msgfrom alibaba").build()); // 3构造一个可订阅的消息通道messageChannel。使用MessageHandler去消费这个消息通道里的消息。发送一条消息到这个消息通道，消息最终被消息通道里的MessageHandler所消费。最后控制台打印出:receive: msg from alibaba。

DirectChannel内部有个UnicastingDispatcher类型的消息分发器，会分发到对应的消息通道MessageChannel中，从名字也可以看出来，UnicastingDispatcher是个单播的分发器，只能选择一个消息通道。那么如何选择呢? 内部提供了LoadBalancingStrategy负载均衡策略，默认只有轮询的实现，可以进行扩展。

我们对上段代码做一点修改，使用多个 MessageHandler 去处理消息：

SubscribableChannel messageChannel = new DirectChannel(); messageChannel.subscribe(msg -> { System.out.println("receive1: " + msg.getPayload()); }); messageChannel.subscribe(msg -> { System.out.println("receive2: " + msg.getPayload()); }); messageChannel.send(MessageBuilder.withPayload("msg from alibaba").build()); messageChannel.send(MessageBuilder.withPayload("msg from alibaba").build());

由于DirectChannel内部的消息分发器是UnicastingDispatcher单播的方式，并且采用轮询的负载均衡策略，所以这里两次的消费分别对应这两个MessageHandler。控制台打印出：

receive1: msg from alibaba receive2: msg from alibaba

既然存在单播的消息分发器UnicastingDispatcher，必然也会存在广播的消息分发器，那就是BroadcastingDispatcher，它被 PublishSubscribeChannel 这个消息通道所使用。广播消息分发器会把消息分发给所有的 MessageHandler：

SubscribableChannel messageChannel = new PublishSubscribeChannel(); messageChannel.subscribe(msg -> { System.out.println("receive1: " + msg.getPayload()); }); messageChannel.subscribe(msg -> { System.out.println("receive2: " + msg.getPayload()); }); messageChannel.send(MessageBuilder.withPayload("msg from alibaba").build()); messageChannel.send(MessageBuilder.withPayload("msg from alibaba").build());Spring Cloud StreamSCS 与各模块之间的关系是：SCS 在 Spring Integration 的基础上进行了封装，提出了Binder, Binding, @EnableBinding, @StreamListener等概念。SCS 与 Spring Boot Actuator 整合，提供了/bindings, /channelsendpoint。SCS 与 Spring Boot Externalized Configuration 整合，提供了BindingProperties, BinderProperties等外部化配置类。SCS 增强了消息发送失败的和消费失败情况下的处理逻辑等功能。SCS 是 Spring Integration 的加强，同时与 Spring Boot 体系进行了融合，也是 Spring Cloud Bus 的基础。它屏蔽了底层消息中间件的实现细节，希望以统一的一套 API 来进行消息的发送/消费，底层消息中间件的实现细节由各消息中间件的 Binder 完成。

Binder是提供与外部消息中间件集成的组件，为构造Binding提供了 2 个方法，分别是bindConsumer和bindProducer，它们分别用于构造生产者和消费者。目前官方的实现有 Rabbit Binder 和 Kafka Binder， Spring Cloud Alibaba 内部已经实现了 RocketMQ Binder。

从图中可以看出，Binding是连接应用程序跟消息中间件的桥梁，用于消息的消费和生产。我们来看一个最简单的使用 RocketMQ Binder 的例子，然后分析一下它的底层处理原理：

启动类及消息的发送：@SpringBootApplication @EnableBinding({ Source.class, Sink.class }) // 1 public class SendAndReceiveApplication { public static void main(String[] args) { SpringApplication.run(SendAndReceiveApplication.class, args); } @Bean // 2 public CustomRunner customRunner() { return new CustomRunner(); } public static class CustomRunner implements CommandLineRunner { @Autowired private Source source; @Override public void run(String... args) throws Exception { int count = 5; for (int index = 1; index

【本文地址】

公司简介

联系我们